package org.sn.jdish.spark.sql;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.DataFrameReader;
import org.apache.spark.sql.DataFrameWriter;
import org.apache.spark.sql.SQLContext;

/**
 * LoadSave
 * 
 * @author snzigod@hotmail.com
 */
public class LoadSave {

	public static void main(String[] args) {
		/**
		 * SparkConf:第一步创建一个SparkConf，在这个对象里面可以设置允许模式Local Standalone yarn
		 * AppName(可以在Web UI中看到) 还可以设置Spark运行时的资源要求
		 */
		SparkConf sparkConf = new SparkConf().setAppName("sparkSql").setMaster("local");

		/**
		 * 基于SparkConf的对象可以创建出来一个SparkContext Spark上下文
		 * SparkContext是通往集群的唯一通道，SparkContext在创建的时候还会创建任务调度器
		 */
		JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);

		readJson(javaSparkContext);

		javaSparkContext.close();
	}

	public static void readJson(JavaSparkContext javaSparkContext) {
		// 创建DataFrame 读取json
		SQLContext sqlContext = new SQLContext(javaSparkContext);

		DataFrameReader dataFrameReader = sqlContext.read();

		// parquet 是本地数据存储的格式
		DataFrame df = dataFrameReader.load("data/users.parquet");

		df.printSchema();
		df.show();

		// 通过关writer写入并保存save
		DataFrameWriter write = df.select("name", "favorite_color").write();
		write.save("namesAndColors.parquet");

		// DataFrame df1 = dataFrameReader
		// .load("namesAndColors.parquet/part-r-00000-7de940b5-233f-43b5-8e4f-1ef80390f28b.gz.parquet");
		// df1.show();
	}
}
